package com.atguigu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.app.BaseAppV1;
import com.atguigu.realtime.bean.CourseReviewBean;
import com.atguigu.realtime.bean.TrafficPageViewBean;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.util.AtguiguUtil;
import com.atguigu.realtime.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @auther yuzehua
 * @date 2022/10/28 - 9:14
 */
public class Dws_CourseReview extends BaseAppV1 {
    public static void main(String[] args) {
        new Dws_CourseReview().init(
                2060,
                2,
                "Dws_CourseReview",
                Constant.TOPIC_DWD_REVIEW_INFO
        );
    }
    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        // 1.封装到一个POJO中
        SingleOutputStreamOperator<CourseReviewBean> beanStream = parseToPojo(stream);
        // 2.开窗聚合
        SingleOutputStreamOperator<CourseReviewBean> resultStream = windowAndAgg(beanStream);
        // 3. 写出到 clickhouse 中
        writeToClickhouse(resultStream);
    }

    private void writeToClickhouse(SingleOutputStreamOperator<CourseReviewBean> resultStream) {
        resultStream
                .addSink(FlinkSinkUtil.getClickHouseSink("dws_course_user_review_view_window",
                        CourseReviewBean.class
                ));
    }

    private SingleOutputStreamOperator<CourseReviewBean> windowAndAgg(SingleOutputStreamOperator<CourseReviewBean> beanStream) {
        return beanStream
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<CourseReviewBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(new SerializableTimestampAssigner<CourseReviewBean>() {
                            @Override
                            public long extractTimestamp(CourseReviewBean element, long recordTimestamp) {
                                return element.getTs();
                            }
                        })
                )
                .keyBy(bean -> bean.getCourseId() + bean.getUserId())
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(new ReduceFunction<CourseReviewBean>() {
                    @Override
                    public CourseReviewBean reduce(CourseReviewBean value1, CourseReviewBean value2) throws Exception {
                        value1.setReviewCount(value1.getReviewCount() + value2.getReviewCount());
                        value1.setReviewStars(value1.getReviewStars() + value2.getReviewStars());
                        return value1;
                    }
                }, new ProcessWindowFunction<CourseReviewBean, CourseReviewBean, String, TimeWindow>() {
                    @Override
                    public void process(String s, Context ctx, Iterable<CourseReviewBean> elements, Collector<CourseReviewBean> out) throws Exception {
                        CourseReviewBean bean = elements.iterator().next();
                        bean.setReviewAvgScore(bean.getReviewStars() / bean.getReviewCount());

                        // 补充窗口开始和结束
                        bean.setStt(AtguiguUtil.toDatTime(ctx.window().getStart()));
                        bean.setEdt(AtguiguUtil.toDatTime(ctx.window().getEnd()));

                        // 表示这个结果的统计时间
                        bean.setTs(System.currentTimeMillis());
                        out.collect(bean);

                    }
                });
    }

    private SingleOutputStreamOperator<CourseReviewBean> parseToPojo(DataStreamSource<String> stream) {
        return stream
                .map(JSON::parseObject)
                .map(new MapFunction<JSONObject, CourseReviewBean>() {
                    @Override
                    public CourseReviewBean map(JSONObject obj) throws Exception {
                        String courseId = obj.getString("course_id");
                        Long reviewStars = obj.getLong("review_stars");
                        String userId = obj.getString("user_id");
                        Long ts = AtguiguUtil.toShootTime(obj.getString("create_time"));

                        // 用户评价次数，来一条数据，评价次数+1
                        Long reviewCount = 1L;

                        return new CourseReviewBean(
                                "","",courseId,userId,reviewStars,reviewCount,
                                0L, ts
                        );
                    }
                });

    }
}
/*

用户平均评分（平均每个用户的评分）
    每来一条数据，按用户分组， 总评分/评价次数

评价用户数 （参与评价的用户数）

好评率 （好评用户数 / 参加评价的用户数）
    5为好评

 */
